package redisv9ex

import (
	"context"
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/sputils"
	"github.com/redis/go-redis/v9"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type RedisSpClient struct {
	startedflag int32

	startlk sync.RWMutex

	ctx      context.Context
	cm       *redis.Client
	pub      *redis.PubSub
	clientId string
	addr     string
	user     string
	pass     string

	ex      *sputils.Subscribe
	closewg sync.WaitGroup

	statusText string
}

var (
	redisClientN int32
)

func (this *RedisSpClient) startWork(psub *redis.PubSub) {
	defer func() {
		this.statusText = fmt.Sprintf("[%s]终止工作", gobase.NowString())
		this.closewg.Done()
	}()

	for msg := range psub.Channel() {
		buf := []byte(msg.Payload)
		this.ex.Pub(msg.Channel, 0, sputils.SpTypeData, buf, msg.Channel)
	}
}

func (this *RedisSpClient) checkConnect() error {
	redisOpt := &redis.Options{
		Addr:     this.addr,
		Password: this.pass,
		Username: this.user,
	}

	this.cm = redis.NewClient(redisOpt)
	ctx := context.Background()
	this.pub = this.cm.PSubscribe(ctx)
	return nil
}

func (this *RedisSpClient) checkPing() {
	go func() {
		var t0 time.Time
		time.Sleep(time.Millisecond * 100)
		for {
			this.startlk.RLock()
			flag := this.startedflag
			this.startlk.RUnlock()
			if flag == 0 {
				break
			}
			if time.Since(t0) > time.Second*60 {
				t0 = time.Now()
				err := this.Ping()
				if err != nil {
					this.statusText = fmt.Sprintf("[%s]Ping err:%s", gobase.NowString(), err.Error())
				} else {
					this.statusText = fmt.Sprintf("[%s]Ping ok", gobase.NowString())
				}
			}
			time.Sleep(time.Millisecond * 500)
		}
	}()
}

func (this *RedisSpClient) CheckStart() error {
	this.startlk.Lock()
	defer this.startlk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()

	this.statusText = fmt.Sprintf("[%s]初始化RedisClient", gobase.NowString())
	err := this.checkConnect()
	if err != nil {
		this.statusText = fmt.Sprintf("[%s]初始化RedisClient异常:%s", gobase.NowString(), err.Error())
		this.ex.Pub(sputils.SpEvtNotify, 0, sputils.SpTypeError, err.Error())
	} else {
		this.closewg.Add(1)
		go this.startWork(this.pub)
		this.ex.Pub(sputils.SpEvtNotify, 0, sputils.SpTypeStart)
		this.checkPing()
	}
	return err
}

func (this *RedisSpClient) closeRedis() {
	if this.cm != nil {
		err := this.cm.Close()
		if err != nil {
			this.statusText = fmt.Sprintf("[%s]停止出现异常:%s", gobase.NowString(), err.Error())
		}
		this.cm = nil
	}
}

func (this *RedisSpClient) Ping() error {
	clt := this.cm
	if clt == nil {
		return gobase.ErrInvalidObject
	}
	_, err := clt.Ping(context.Background()).Result()
	return err
}

func (this *RedisSpClient) Close() error {
	this.startlk.Lock()
	defer this.startlk.Unlock()
	if this.startedflag == 1 {
		this.startedflag = 0
		this.closeRedis()
	}
	atomic.AddInt32(&redisClientN, -1)
	return nil
}

func (this *RedisSpClient) WaitDone() {
	this.closewg.Wait()
}

// 1: topic, args[0].([]byte/string)
// 1: topic, args[0].([]byte/string), args[1].(pubArg)
// 没办法获取到发送成功的数量, 1:表示发送成功了
func (this *RedisSpClient) Pub(topic string, max int, args ...interface{}) int {
	clt := this.cm
	if clt == nil {
		this.ex.Pub(sputils.SpEvtNotify, 0, sputils.SpTypeError, fmt.Errorf("pub:%s, err:%s", topic, "redis is nil"))
		return -1
	}

	var data []byte
	if buf, ok := args[0].([]byte); ok {
		data = buf
	} else if str, ok := args[0].(string); ok {
		data = []byte(str)
	} else {
		return -1
	}

	pubctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	r := clt.Publish(pubctx, topic, data)
	n, err := r.Result()
	if err != nil {
		this.ex.Pub(sputils.SpEvtNotify, 0, sputils.SpTypeError, fmt.Errorf("pub:%s, err:%s", topic, err.Error()))
	} else {
		this.ex.Pub(sputils.SpEvtNotifyExt, 0, sputils.SpTypeSent, len(data), data, topic)
	}
	return int(n)
}

func (this *RedisSpClient) Sub(id, channel string, cb subpub.SubFunc) {
	r := this.ex.Sub(id, channel, cb)
	if r == 1 {
		if strings.HasPrefix(channel, sputils.SpEvtPrefix) {
			return
		}
		this.pub.PSubscribe(context.Background(), channel)
	}
}

func (this *RedisSpClient) Unsub(id, channel string) bool {
	r, _ := this.ex.Unsub(id, channel)
	if r == 0 {
		if strings.HasPrefix(channel, sputils.SpEvtPrefix) {
			return true
		}
		this.pub.Unsubscribe(context.Background(), channel)
	}
	return true
}

func (this *RedisSpClient) GetStatus(args ...interface{}) string {
	return fmt.Sprintf("%s,%s,ex:%s", this.addr, this.statusText, this.ex.Status())
}

func (this *RedisSpClient) ConvertArgs(action int, confMap gobase.StrMap) interface{} {

	return nil
}

var (
	DefaultRedisSpClient = gobase.NewConnectObjectRefStorage(newRedisSpClient)
)

// connstr=&user=&pass=&conrand=&topic=
func newRedisSpClient(connstr string) interface{} {
	conf := gobase.NewStrMap()
	conf.URLFormDecode(connstr)
	node := &RedisSpClient{ex: sputils.NewSubscribe()}
	node.addr = conf.StringByName("addr", "")
	if len(node.addr) == 0 {
		node.addr = conf.StringByName("connstr", "")
	}
	node.pass = conf.StringByName("pass", "")
	node.user = conf.StringByName("user", "")
	node.clientId = conf.StringByName("clientid", "")
	atomic.AddInt32(&redisClientN, 1)
	return node
}

func checkGetRedisSpClient(args ...interface{}) (rval interface{}, err error) {
	if len(args) == 0 {
		err = fmt.Errorf("无效的参数")
		return
	}

	if conf, ok := args[0].(gobase.StrMap); ok {
		s := conf.URLFormEncodeKeylst("connstr", "addr", "user", "pass", "clientid", "connrand")
		rval = DefaultRedisSpClient.CheckGetRef(s)
		return
	} else {
		err = fmt.Errorf("无效的参数")
		return
	}
}

func init() {
	// 返回一个objectRef, 返回的objectRef必须实现isubpub,iclose接口
	factory.RegisterCreateFunc("sp.redis", checkGetRedisSpClient)

}
